﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Net;
using System.Net.Sockets;
using System.Text;
using UMC.Data;
using System.Linq;
using System.Threading.Tasks;

namespace UMC.Net
{
    public class NetSubscribe : NetBridge
    {
        String _ip, _Key;
        int _port;

        public string Key => _Key;

        public string Address => _ip;
        public int Port => _port;
        private NetSubscribe()
        {

        }
        public NetSubscribe(String point, String ip, int port, StringValue appSecret)
        {

            this._point = point;
            this._ip = ip;
            this._port = port;
            _Subscribes[this._ip] = this;

            this.appSecret = appSecret;
            Connect();
        }
        StringValue appSecret;
        public async void Connect()
        {
            if (this._port > 0)
            {
                var client = new Socket(SocketType.Stream, ProtocolType.Tcp);

                try
                {

                    await client.ConnectAsync(this._ip, this._port);
                    Connect(client);
                }
                catch
                {

                    await System.Threading.Tasks.Task.Delay(10000);

                    Connect();



                }
            }
        }

        async void Connect(Socket tcpClient)
        {

            var buffers = System.Buffers.ArrayPool<byte>.Shared.Rent(0x200);
            try
            {

                buffers[0] = 0x02;
                buffers[1] = 0x03;
                buffers[2] = 0x01;
                buffers[3] = (byte)this._point.WriteBytes(buffers, 4);
                var size = buffers[1] + 4;

                BitConverter.TryWriteBytes(buffers.AsSpan(size), UMC.Data.Utility.TimeSpan());
                size += 4;
                buffers[3] += 4;
                appSecret.Span.CopyTo(buffers.AsSpan(size));
                // size += appSecret.Span.Length;
                int len = appSecret.Span.Length;//appSecret.WriteBytes(buffers, size);
                Utility.MD5(buffers.AsSpan(4, size - 4 + len), buffers.AsSpan(size + len));
                buffers[size] = 0x16;
                buffers.AsSpan(size + len).CopyTo(buffers.AsSpan(size + 1));
                size += 17;
                buffers[size] = 0x00;
                size++;


                await tcpClient.SendAsync(new ArraySegment<byte>(buffers, 0, size));

                await tcpClient.ReceiveAsync(new ArraySegment<byte>(buffers, 0, 2));
                if (buffers[0] == 0x00)
                {
                    var size1 = await tcpClient.ReceiveAsync(new ArraySegment<byte>(buffers, 0, buffers[1]));

                    this._Key = buffers.AsSpan(0, size1).UTF8();

                    this.Bridge(tcpClient);
                }
                else
                {
                    tcpClient.Close();
                    tcpClient.Dispose();
                    return;
                }
            }
            finally
            {
                System.Buffers.ArrayPool<byte>.Shared.Return(buffers);
            }

        }
        String _point;
        public string Point => _point;
        public override void Close()
        {
            base.Close();
            if (_port > 0)
            {
                if (_Subscribes.ContainsKey(this._ip))
                {
                    System.Threading.Tasks.Task.Factory.StartNew(async delegate
                    {
                        await System.Threading.Tasks.Task.Delay(10000);

                        Connect();

                    });
                }
            }
        }
        public static NetSubscribe[] Subscribes => _Subscribes.Values.ToArray();

        public static NetSubscribe[] Publishes => _Publishes.Values.ToArray();

        public static NetSubscribe Subscribe(String subscribeKey, String ip, String point, Socket stream, StringValue appSecret)
        {

            if (_Publishes.ContainsKey(subscribeKey))
            {
                NetSubscribe slave = _Publishes[subscribeKey];
                slave.Bridge(stream);
                return slave;
            }
            else
            {
                var subscribe = new NetSubscribe
                {
                    _Key = subscribeKey,
                    _point = point,
                    appSecret = appSecret,
                    _ip = ip
                };
                subscribe.Bridge(stream);
                _Publishes[subscribeKey] = subscribe;
                return subscribe;
            }





        }

        protected override void Read(int pid, byte[] buffer, int offset, int count)
        {
            if (pid == 0)
            {
                if (count > 0)
                {
                    var eIndex = UMC.Data.Utility.FindIndex(buffer, offset, offset + count, new byte[] { 46 });
                    if (eIndex > -1)
                    {
                        var len = eIndex - offset;
                        var key = System.Text.Encoding.UTF8.GetString(buffer, offset, len);
                        var message = System.Text.Encoding.UTF8.GetString(buffer, eIndex + 1, count - len - 1);

                        switch (key)
                        {
                            case "Subscribe":
                                switch (message)
                                {
                                    case "Delete":
                                        if (this._port > 0)
                                        {
                                            _Subscribes.Remove(this._ip);

                                            _RemoveConfig();
                                        }
                                        else
                                        {
                                            _Publishes.Remove(this._Key);
                                        }
                                        this.Close();
                                        break;
                                }
                                break;
                            case "Export":
                                var nameCode = UMC.Data.Utility.IntParse(message, 0);
                                if (nameCode != 0)
                                {
                                    var cacheset = HotCache.Caches().FirstOrDefault(r => r.NameCode == nameCode);
                                    if (cacheset != null)
                                    {
                                        System.Threading.Tasks.Task.Factory.StartNew(() => this.Export(cacheset));
                                    }
                                }
                                break;
                            default:

                                _MessageSubscribes[key]?.Subscribe(message);

                                break;
                        }
                    }
                }
            }
            else
            {

                _DataSubscribes[pid]?.Subscribe(buffer, offset, count);

            }
        }

        void Export(UMC.Data.Caches.ICacheSet cacheSet)
        {
            var buffer = System.Buffers.ArrayPool<byte>.Shared.Rent(0x200);// new byte[0x200];
            cacheSet.Export(row =>
            {
                var size = 9;
                foreach (var cell in row)
                {
                    size += cell.Length + 6;
                }
                size += 2;
                if (buffer.Length < size)
                {
                    System.Buffers.ArrayPool<byte>.Shared.Return(buffer);
                    buffer = System.Buffers.ArrayPool<byte>.Shared.Rent(size);//new byte[size];
                }

                buffer[0] = STX;
                Array.Copy(BitConverter.GetBytes(cacheSet.NameCode), 0, buffer, 1, 4);
                Array.Copy(BitConverter.GetBytes(size - 9), 0, buffer, 5, 4);


                var index = 9;
                foreach (var cell in row)
                {
                    byte cindex = cell[0];
                    cindex += 16;
                    buffer[index] = cindex;
                    index++;
                    var l = cell.Length - 1;
                    Array.Copy(BitConverter.GetBytes(l), 0, buffer, index, 4);
                    index += 4;

                    Array.Copy(cell, 1, buffer, index, l);
                    index += l;
                    buffer[index] = 13;
                    index++;
                    buffer[index] = 10;
                    index++;

                }
                buffer[index] = 13;
                buffer[index + 1] = 10;

                this.socket.Send(buffer, 0, size, SocketFlags.None);


            });
            System.Buffers.ArrayPool<byte>.Shared.Return(buffer);

        }

        public static void Publish(String key, String message)
        {
            var pub = System.Text.Encoding.UTF8.GetBytes($"{key}.{message}");
            if (pub.Length > 0)
            {
                Publish(0, pub, 0, pub.Length);
            }

            _MessageSubscribes[key]?.Subscribe(message);
        }

        static Dictionary<String, IStringSubscribe> _MessageSubscribes = new Dictionary<string, IStringSubscribe>();
        static Dictionary<String, NetSubscribe> _Publishes = new Dictionary<string, NetSubscribe>();
        static Dictionary<String, NetSubscribe> _Subscribes = new Dictionary<string, NetSubscribe>();
        static Dictionary<int, IDataSubscribe> _DataSubscribes = new Dictionary<int, IDataSubscribe>();


        public static void Publish(int publishId, byte[] buffer, int index, int length)
        {
            var m = _Publishes.GetEnumerator();

            while (m.MoveNext())
            {
                m.Current.Value.Write(publishId, buffer, index, length);
            }
        }
        public void Import(int nameCode)
        {

            var data = System.Text.Encoding.UTF8.GetBytes($"Export.{nameCode}");

            var size = data.Length + 9;
            var buffer = System.Buffers.ArrayPool<byte>.Shared.Rent(size); //new byte[size];
            buffer[0] = STX;
            Array.Copy(BitConverter.GetBytes(0), 0, buffer, 1, 4);
            Array.Copy(BitConverter.GetBytes(data.Length), 0, buffer, 5, 4);
            Array.Copy(data, 0, buffer, 9, data.Length);
            try
            {
                this.socket.Send(buffer, 0, size, SocketFlags.None);
            }
            catch
            {
                this.Close();
            }
            finally
            {

                System.Buffers.ArrayPool<byte>.Shared.Return(buffer);
            }
        }
        public void Remove()
        {

            var buffer = System.Buffers.ArrayPool<byte>.Shared.Rent(100);
            var l = System.Text.Encoding.UTF8.GetBytes("Subscribe.Delete", buffer.AsSpan(9));

            buffer[0] = STX;
            BitConverter.TryWriteBytes(buffer.AsSpan(1), 0);
            BitConverter.TryWriteBytes(buffer.AsSpan(5), l);

            try
            {
                if (this.IsBridging)
                {
                    this.socket.Send(buffer, 0, l + 9, SocketFlags.None);

                }
            }
            catch
            {
                this.Close();
            }
            finally
            {
                System.Buffers.ArrayPool<byte>.Shared.Return(buffer);
            }
            if (this._port > 0)
            {
                _Subscribes.Remove(this.Address);
                _RemoveConfig();
                base.Close();
            }
            else
            {
                _Publishes.Remove(this.Key);
                this.Close();
            }

        }
        void _RemoveConfig()
        {

            var Subscribe = Reflection.Configuration("Subscribe") ?? new ProviderConfiguration();
            var em = Subscribe.Providers.GetEnumerator();
            while (em.MoveNext())
            {
                var subscr = em.Current;// Subscribe[i];
                var url = subscr.Attributes["url"];
                if (String.IsNullOrEmpty(url) == false)
                {
                    if (url.Contains(this._ip))
                    {
                        Subscribe.Remove(subscr.Name);
                        break;
                    }
                }

            }
            Subscribe.WriteTo(Reflection.AppDataPath("UMC//Subscribe.xml"));
        }

        public static void Subscribe(String key, IStringSubscribe subscriber)
        {
            _MessageSubscribes[key] = subscriber;
        }

        public static void Subscribe(int nameCode, IDataSubscribe subscriber)
        {
            _DataSubscribes[nameCode] = subscriber;
        }

    }

}
